/*-------------------------------------------------------------------------
 *
 * logicalproto.h
 *		逻辑复制协议
 *
 * Copyright (c) 2015-2022, PostgreSQL Global Development Group
 *
 * IDENTIFICATION
 *		src/include/replication/logicalproto.h
 *
 *-------------------------------------------------------------------------
 */
#ifndef LOGICAL_PROTO_H
#define LOGICAL_PROTO_H

#include "access/xact.h"
#include "executor/tuptable.h"
#include "replication/reorderbuffer.h"
#include "utils/rel.h"

/*
 * 协议能力
 *
 * LOGICALREP_PROTO_VERSION_NUM是我们的本地协议。
 * LOGICALREP_PROTO_MAX_VERSION_NUM是我们可以支持的最大版本。
 * LOGICALREP_PROTO_MIN_VERSION_NUM是我们有向后兼容性的最旧版本。客户端在连接时请求协议版本。
 *
 * LOGICALREP_PROTO_STREAM_VERSION_NUM是支持流式大事务的最小协议版本。于PG14引入。
 *
 * LOGICALREP_PROTO_TWOPHASE_VERSION_NUM是支持两阶段提交解码（在准备时）的最小协议版本。于PG15引入。
 */
#define LOGICALREP_PROTO_MIN_VERSION_NUM 1
#define LOGICALREP_PROTO_VERSION_NUM 1
#define LOGICALREP_PROTO_STREAM_VERSION_NUM 2
#define LOGICALREP_PROTO_TWOPHASE_VERSION_NUM 3
#define LOGICALREP_PROTO_MAX_VERSION_NUM LOGICALREP_PROTO_TWOPHASE_VERSION_NUM

/*
 * 逻辑消息类型
 *
 * 由逻辑复制wire协议使用。
 *
 * 注意：尽管这是一个枚举，但这些值用于在逻辑复制协议中识别消息类型，
 * 该协议使用一个字节来识别消息类型。因此，这些值应为单字节宽，
 * 并且最好是可读的字符。
 */
typedef enum LogicalRepMsgType
{
	LOGICAL_REP_MSG_BEGIN = 'B',
	LOGICAL_REP_MSG_COMMIT = 'C',
	LOGICAL_REP_MSG_ORIGIN = 'O',
	LOGICAL_REP_MSG_INSERT = 'I',
	LOGICAL_REP_MSG_UPDATE = 'U',
	LOGICAL_REP_MSG_DELETE = 'D',
	LOGICAL_REP_MSG_TRUNCATE = 'T',
	LOGICAL_REP_MSG_RELATION = 'R',
	LOGICAL_REP_MSG_TYPE = 'Y',
	LOGICAL_REP_MSG_MESSAGE = 'M',
	LOGICAL_REP_MSG_BEGIN_PREPARE = 'b',
	LOGICAL_REP_MSG_PREPARE = 'P',
	LOGICAL_REP_MSG_COMMIT_PREPARED = 'K',
	LOGICAL_REP_MSG_ROLLBACK_PREPARED = 'r',
	LOGICAL_REP_MSG_STREAM_START = 'S',
	LOGICAL_REP_MSG_STREAM_STOP = 'E',
	LOGICAL_REP_MSG_STREAM_COMMIT = 'c',
	LOGICAL_REP_MSG_STREAM_ABORT = 'A',
	LOGICAL_REP_MSG_STREAM_PREPARE = 'p'
} LogicalRepMsgType;

/*
 * 此结构存储通过逻辑复制接收到的元组。
 * 请记住，这些列对应于*远程*表。
 */
typedef struct LogicalRepTupleData
{
	/* 每列一个的StringInfos数组；某些可能未被使用 */
	StringInfoData *colvalues;
	/* 数组的标记，用于表示空/未更改/文本/二进制，每列一个 */
	char	   *colstatus;
	/* 上述数组的长度 */
	int			ncols;
} LogicalRepTupleData;

/* LogicalRepTupleData.colstatus[colnum] 的可能值 */
/* 这些值也用于网络协议中 */
#define LOGICALREP_COLUMN_NULL		'n'
#define LOGICALREP_COLUMN_UNCHANGED	'u'
#define LOGICALREP_COLUMN_TEXT		't'
#define LOGICALREP_COLUMN_BINARY	'b' /* 在 PG14 中新增 */

typedef uint32 LogicalRepRelId;

/* 关系信息 */
typedef struct LogicalRepRelation
{
	/* 来自远程端的信息。 */
	LogicalRepRelId remoteid;	/* 关系的唯一 ID */
	char	   *nspname;		/* 模式名称 */
	char	   *relname;		/* 关系名称 */
	int			natts;			/* 列数 */
	char	  **attnames;		/* 列名 */
	Oid		   *atttyps;		/* 列类型 */
	char		replident;		/* 副本身份 */
	char		relkind;		/* 远程关系种类 */
	Bitmapset  *attkeys;		/* 关键列的位图 */
} LogicalRepRelation;

/* 类型映射信息 */
typedef struct LogicalRepTyp
{
	Oid			remoteid;		/* 远程类型的唯一 ID */
	char	   *nspname;		/* 远程类型的模式名称 */
	char	   *typname;		/* 远程类型的名称 */
} LogicalRepTyp;

/* 事务信息 */
typedef struct LogicalRepBeginData
{
	XLogRecPtr	final_lsn;
	TimestampTz committime;
	TransactionId xid;
} LogicalRepBeginData;

typedef struct LogicalRepCommitData
{
	XLogRecPtr	commit_lsn;
	XLogRecPtr	end_lsn;
	TimestampTz committime;
} LogicalRepCommitData;

/*
 * 准备事务协议信息，用于开始准备和准备。
 */
typedef struct LogicalRepPreparedTxnData
{
	XLogRecPtr	prepare_lsn;
	XLogRecPtr	end_lsn;
	TimestampTz prepare_time;
	TransactionId xid;
	char		gid[GIDSIZE];
} LogicalRepPreparedTxnData;

/*
 * 准备事务协议信息，用于提交准备好的事务。
 */
typedef struct LogicalRepCommitPreparedTxnData
{
	XLogRecPtr	commit_lsn;
	XLogRecPtr	end_lsn;
	TimestampTz commit_time;
	TransactionId xid;
	char		gid[GIDSIZE];
} LogicalRepCommitPreparedTxnData;

/*
 * 回滚准备事务协议信息。prepare的信息
 * prepare_end_lsn 和 prepare_time 用于检查下游是否
 * 已接收此准备事务，在这种情况下可以应用回滚，
 * 否则，可以跳过回滚操作。仅凭 gid 是不够的，
 * 因为下游节点可能有具有相同标识符的准备事务。
 */
typedef struct LogicalRepRollbackPreparedTxnData
{
	XLogRecPtr	prepare_end_lsn;
	XLogRecPtr	rollback_end_lsn;
	TimestampTz prepare_time;
	TimestampTz rollback_time;
	TransactionId xid;
	char		gid[GIDSIZE];
} LogicalRepRollbackPreparedTxnData;

extern void logicalrep_write_begin(StringInfo out, ReorderBufferTXN *txn);
extern void logicalrep_read_begin(StringInfo in,
								  LogicalRepBeginData *begin_data);
extern void logicalrep_write_commit(StringInfo out, ReorderBufferTXN *txn,
									XLogRecPtr commit_lsn);
extern void logicalrep_read_commit(StringInfo in,
								   LogicalRepCommitData *commit_data);
extern void logicalrep_write_begin_prepare(StringInfo out, ReorderBufferTXN *txn);
extern void logicalrep_read_begin_prepare(StringInfo in,
										  LogicalRepPreparedTxnData *begin_data);
extern void logicalrep_write_prepare(StringInfo out, ReorderBufferTXN *txn,
									 XLogRecPtr prepare_lsn);
extern void logicalrep_read_prepare(StringInfo in,
									LogicalRepPreparedTxnData *prepare_data);
extern void logicalrep_write_commit_prepared(StringInfo out, ReorderBufferTXN *txn,
											 XLogRecPtr commit_lsn);
extern void logicalrep_read_commit_prepared(StringInfo in,
											LogicalRepCommitPreparedTxnData *prepare_data);
extern void logicalrep_write_rollback_prepared(StringInfo out, ReorderBufferTXN *txn,
											   XLogRecPtr prepare_end_lsn,
											   TimestampTz prepare_time);
extern void logicalrep_read_rollback_prepared(StringInfo in,
											  LogicalRepRollbackPreparedTxnData *rollback_data);
extern void logicalrep_write_stream_prepare(StringInfo out, ReorderBufferTXN *txn,
											XLogRecPtr prepare_lsn);
extern void logicalrep_read_stream_prepare(StringInfo in,
										   LogicalRepPreparedTxnData *prepare_data);

extern void logicalrep_write_origin(StringInfo out, const char *origin,
									XLogRecPtr origin_lsn);
extern char *logicalrep_read_origin(StringInfo in, XLogRecPtr *origin_lsn);
extern void logicalrep_write_insert(StringInfo out, TransactionId xid,
									Relation rel,
									TupleTableSlot *newslot,
									bool binary, Bitmapset *columns);
extern LogicalRepRelId logicalrep_read_insert(StringInfo in, LogicalRepTupleData *newtup);
extern void logicalrep_write_update(StringInfo out, TransactionId xid,
									Relation rel,
									TupleTableSlot *oldslot,
									TupleTableSlot *newslot, bool binary, Bitmapset *columns);
extern LogicalRepRelId logicalrep_read_update(StringInfo in,
											  bool *has_oldtuple, LogicalRepTupleData *oldtup,
											  LogicalRepTupleData *newtup);
extern void logicalrep_write_delete(StringInfo out, TransactionId xid,
									Relation rel, TupleTableSlot *oldtuple,
									bool binary, Bitmapset *columns);
extern LogicalRepRelId logicalrep_read_delete(StringInfo in,
											  LogicalRepTupleData *oldtup);
extern void logicalrep_write_truncate(StringInfo out, TransactionId xid,
									  int nrelids, Oid relids[],
									  bool cascade, bool restart_seqs);
extern List *logicalrep_read_truncate(StringInfo in,
									  bool *cascade, bool *restart_seqs);
extern void logicalrep_write_message(StringInfo out, TransactionId xid, XLogRecPtr lsn,
									 bool transactional, const char *prefix, Size sz, const char *message);
extern void logicalrep_write_rel(StringInfo out, TransactionId xid,
								 Relation rel, Bitmapset *columns);
extern LogicalRepRelation *logicalrep_read_rel(StringInfo in);
extern void logicalrep_write_typ(StringInfo out, TransactionId xid,
								 Oid typoid);
extern void logicalrep_read_typ(StringInfo out, LogicalRepTyp *ltyp);
extern void logicalrep_write_stream_start(StringInfo out, TransactionId xid,
										  bool first_segment);
extern TransactionId logicalrep_read_stream_start(StringInfo in,
												  bool *first_segment);
extern void logicalrep_write_stream_stop(StringInfo out);
extern void logicalrep_write_stream_commit(StringInfo out, ReorderBufferTXN *txn,
										   XLogRecPtr commit_lsn);
extern TransactionId logicalrep_read_stream_commit(StringInfo out,
												   LogicalRepCommitData *commit_data);
extern void logicalrep_write_stream_abort(StringInfo out, TransactionId xid,
										  TransactionId subxid);
extern void logicalrep_read_stream_abort(StringInfo in, TransactionId *xid,
										 TransactionId *subxid);
extern const char *logicalrep_message_type(LogicalRepMsgType action);

#endif							/* LOGICAL_PROTO_H */
